package com.motu.monstercity.server.gate;

import com.motu.vertx.module.utility.base.ServerConfig;
import com.motu.vertx.module.utility.kafka.KafkaManager;
import com.motu.vertx.module.utility.toolset.Tool;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonArray;
import io.vertx.kafka.client.producer.KafkaProducer;
import io.vertx.kafka.client.producer.KafkaProducerRecord;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class ProtocolKafkaManager {

    public final static Logger logger = LoggerFactory.getLogger("ProtocolKafkaManager");
    public static KafkaProducer<String, String> log_producer;// log库的生产者

    public static void init(Vertx vertx) throws Exception {


        // 创建log kafka
        if (ServerConfig.getAppConfig().containsKey("log-kafka-address")) {
            Properties log_config = new Properties();
            log_config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ServerConfig.getAppConfig().getString("log-kafka-address"));
            log_config.put("key.serializer", StringSerializer.class);
            log_config.put("value.serializer", StringSerializer.class);
            log_config.put("acks", "1");
            // 优化kafka生产者配置
            KafkaManager.upgradeProducerConfig(log_config, ServerConfig.getAppConfig());
            log_producer = KafkaProducer.create(vertx, log_config);
        }
    }

    public static void enterClickhouseQueue(JsonArray params) {
        KafkaProducerRecord<String, String> record = KafkaProducerRecord.create("protocol_log", params.toString());
        KafkaProducer<String, String> kafkaProducer;
        kafkaProducer = log_producer;
        kafkaProducer.send(record, done -> {
            if (done.failed()) {
                logger.error("protocol pushClickhouseDBQueue fail cause = " + Tool.getException(done.cause()));
            } else {
//                RecordMetadata recordMetadata = done.result();
//                MainVerticle.logger.info("protocol pushClickhouseDBQueue Message " + record.value().toString() + " written on topic=" + recordMetadata.getTopic() +
//                        ", partition=" + recordMetadata.getPartition() +
//                        ", offset=" + recordMetadata.getOffset());
            }
        });

    }
}
